package zjs.dc.controller;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * @Class:
 * @Author: kongcb
 * @Description: TODO
 * @Date: 2021/12/15 9:23
 */
public class TImportExport {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("t_import_export").enableHiveSupport().getOrCreate();
        spark.sparkContext().setLogLevel("ERROR");
        Dataset<Row> zs_data = spark.sql("select * from a_dc_ep_ods.zs_t_customs_register");
        long zs_count = zs_data.count();
        if(zs_count>0){

            //企查查进出口表数据取统一社会信用代码
            spark.sql("select a.*,b.credit_code from a_dc_ep_dwi.t_import_export a left join a_dc_ep_ods.t_eci_company b on a.company_id = b.company_id")
                    .createOrReplaceTempView("qcc_import_export");
            //新增
            spark.sql("insert overwrite table a_dc_ep_incr.t_import_export select " +
                    "zs.RECORD_ID," +
                    "qcc.key_no," +
                    "qcc.company_id," +
                    "qcc.company_name," +
                    "zs.RECORD_ID," +
                    "zs.GACCID," +
                    "zs.REGGACC," +
                    "if(zs.TRADETYPE='进出口收发货人','进出口货物收发货人',zs.TRADETYPE)," +
                    "zs.REGDATE," +
                    "zs.jobid," +
                    "'1' as isadd" +
                    " from a_dc_ep_ods.zs_t_customs_register zs left join (select a.* from a_dc_ep_ods.t_eci_company a where a.credit_code is not null and a.credit_code !='') qcc on upper(trim(zs.credit_code)) = upper(trim(qcc.CREDIT_CODE)) where qcc.CREDIT_CODE is not null");
            spark.sql("insert into table a_dc_ep_incr.t_import_export select " +
                    "zs.RECORD_ID," +
                    "qcc.key_no," +
                    "qcc.company_id," +
                    "qcc.company_name," +
                    "zs.RECORD_ID," +
                    "zs.GACCID," +
                    "zs.REGGACC," +
                    "if(zs.TRADETYPE='进出口收发货人','进出口货物收发货人',zs.TRADETYPE)," +
                    "zs.REGDATE," +
                    "zs.jobid," +
                    "'1' as isadd" +
                    " from a_dc_ep_ods.zs_t_customs_register zs left join (select a.* from a_dc_ep_ods.t_eci_company a where a.credit_code is null or a.credit_code ='') qcc on trim(zs.entname) = trim(qcc.company_name) where qcc.company_name is not null");
            //删除
            spark.sql("insert into table a_dc_ep_incr.t_import_export select " +
                    "qcc.id,\n" +
                    "qcc.key_no,\n" +
                    "qcc.company_id,\n" +
                    "qcc.company_name,\n" +
                    "qcc.ie_id,\n" +
                    "qcc.reg_no,\n" +
                    "qcc.reg_gov,\n" +
                    "qcc.trade_type,\n" +
                    "qcc.reg_date," +
                    "qcc.dates," +
                    "'-1' as isadd " +
                    " from qcc_import_export qcc inner join (select del.* from a_dc_ep_ods.zs_t_customs_register_del del where del.creditcode is not null and del.creditcode !='') zs on upper(trim(qcc.credit_code)) = upper(trim(zs.CREDITCODE))");

        }else {
            spark.sql("TRUNCATE TABLE a_dc_ep_incr.t_import_export");
            System.out.println("中数本期数据涉及该表的数据为空。。。。。。");
        }
        spark.stop();
    }
}
